package com.dapeng.reactor;

import com.dapeng.debug.Log;
import lombok.Getter;

import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class SelectorThread extends ThreadLocal<LinkedBlockingQueue<Channel>> implements Runnable, Comparable<SelectorThread> {


    @Getter
    private Selector selector;

    private SelectorGroup group;

    private LinkedBlockingQueue<Channel> channelQueue = get();

    @Override
    protected LinkedBlockingQueue<Channel> initialValue() {
        return new LinkedBlockingQueue<>();
    }

    public void addChannelQueue(Channel channel) {
        channelQueue.add(channel);
    }

    public int getActiveChannelNum() {
        return channelQueue.size();
    }

    public SelectorThread(SelectorGroup group) {
        try {
            selector = Selector.open();
            this.group = group;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void run() {
        while (true) {
            try {
                Log.info("{} before ==== {}", Thread.currentThread().getName(), getSelector().keys().size());
                int num = selector.select();
                Log.info("{} after ==== {}", Thread.currentThread().getName(), getSelector().keys().size());
                if (num > 0) {
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = keys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        if (selectionKey.isAcceptable()) {
                            accessHandle(selectionKey);
                        } else if (selectionKey.isReadable()) {
                            readHandle(selectionKey);
                        } else if (selectionKey.isWritable()) {

                        }
                    }
                }

                // 处理其他任务
                while (!channelQueue.isEmpty()) {
                    Channel channel = channelQueue.take();
                    if (channel instanceof ServerSocketChannel) {
                        ((ServerSocketChannel) channel).register(selector, SelectionKey.OP_ACCEPT);
                    } else if (channel instanceof SocketChannel) {
                        ByteBuffer byteBuffer = ByteBuffer.allocateDirect(4096);
                        ((SocketChannel) channel).register(selector, SelectionKey.OP_READ, byteBuffer);
                    }
                }
            } catch (Exception e) {
                Log.error(e);
            }
        }
    }

    private void readHandle(SelectionKey selectionKey) throws Exception {
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();
        byteBuffer.clear();

        while (true) {
            int byn = socketChannel.read(byteBuffer);
            if (byn > 0) {
                byteBuffer.flip();
                while (byteBuffer.hasRemaining()) {
                    socketChannel.write(byteBuffer);
                }
                byteBuffer.clear();
            } else if (byn == 0) {
                break;
            } else {
                Log.info("客户端退出:{}", socketChannel.getRemoteAddress());
                selectionKey.channel();
                socketChannel.close();
                break;
            }
        }
    }

    private void accessHandle(SelectionKey selectionKey) throws Exception {
        ServerSocketChannel serverChannel = (ServerSocketChannel) selectionKey.channel();
        SocketChannel clientChannel = serverChannel.accept();
        clientChannel.configureBlocking(false);

        // 注册到selector
        this.group.registerSelect(clientChannel);
    }

    @Override
    public int compareTo(SelectorThread o) {
        return this.selector.keys().size() - o.selector.keys().size();
    }
}
